package com.jxb.broker;

import com.jxb.RemotingService;
import com.jxb.ThreadFactoryImpl;
import com.jxb.broker.store.CommitLog;
import com.jxb.broker.store.ConsumeQueue;
import com.jxb.netty.*;
import io.netty.channel.ChannelHandlerContext;

import java.io.File;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class BrokerStartup {

    private static final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
            "BrokerControllerScheduledThread"));

    private static int offset;

    private static RemotingService remotingService = new RemotingClient();

    private static final ConcurrentHashMap<String,byte[]> commitLogBuffer = new ConcurrentHashMap<>();

    private static ConcurrentHashMap<String, List<ConsumeQueue>> consumeQueues = new ConcurrentHashMap<String, List<ConsumeQueue>>();

    public static void main(String[] args) throws InterruptedException {
        String clientName = "127.0.0.1";
        String[] namesrvAddrs = {"127.0.0.1:9876","127.0.0.1:9875"};

        //启动namesrv的客户端，连接namesrvs
        remotingService.contectNamesrvs(namesrvAddrs,clientName);

        //定时注册broker
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                registerBrokerAll(remotingService);
            }
        },10,30, TimeUnit.SECONDS);

        //启动broker服务端
        NettyRemotingServer server = new NettyRemotingServer(10911, "broker server");
        SendMessageProcessor sendMessageProcessor = new SendMessageProcessor();
        ExecutorService executorService = Executors.newFixedThreadPool(8, new ThreadFactoryImpl("RemotingExecutorThread_"));
        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(sendMessageProcessor,executorService);
        server.defaultRequestProcessor = pair;
        server.start();
    }

    private static void registerBrokerAll(RemotingService remotingService) {
        final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
        requestHeader.setBrokerAddr("127.0.0.1");
        requestHeader.setBrokerId(1L);
        requestHeader.setBrokerName("broker1");
        requestHeader.setClusterName("cluster1");
        requestHeader.setHaServerAddr("127.0.0.1");
        requestHeader.setCompressed(false);
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
        String body = "127.0.0.1 broker1";
        request.setBody(body.getBytes(StandardCharsets.UTF_8));
        remotingService.sendOneway(request);
    }

    public static class SendMessageProcessor implements NettyRequestProcessor {

        @Override
        public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
            SendMessageRequestHeader requestHeader =
                    (com.jxb.netty.SendMessageRequestHeader) request.decodeCommandCustomHeader(SendMessageRequestHeader.class);
            String topic = requestHeader.getTopic();
            Long code = Long.valueOf(request.getCode());
            if (topic != null && request.getBody() != null) {
                //追加写入缓存
                commitLogBuffer.put(topic,request.getBody());
                //异步通知刷入磁盘
                asyncFile(topic,code);
            }
            return null;
        }

        @Override
        public boolean rejectRequest() {
            return false;
        }
    }

    private static void asyncFile(String topic, Long code) {
        int startPosition = offset;
        //commitlog-待异步处理
        int position = CommitLog.flush(offset,commitLogBuffer.get(topic));
        offset = position;
        //consumequeue-待异步持久化处理
        consumeQueueFlush(topic,code,startPosition);
        //index-存的是消息开始的时间、结束的时间、开始的物理地址和结束的物理地址-待缓存异步持久化处理

    }

    private static void consumeQueueFlush(String topic, Long code, int startPosition) {
        ConsumeQueue consumeQueue = new ConsumeQueue();
        consumeQueue.setMsgId(code);
        consumeQueue.setOffset(startPosition);
        consumeQueue.setTopic(topic);
        if (consumeQueues.containsKey(topic)) {
            List<ConsumeQueue> list = consumeQueues.get(topic);
            if (!list.isEmpty()) {
                list.add(consumeQueue);
                consumeQueues.put(topic,list);
            }
        }else {
            consumeQueues.put(topic,new ArrayList<>());
        }
    }

}
